package kafka

import (
	"errors"
	"github.com/Shopify/sarama"
)

// CreateTopicOptions holds info about topic configuration
type CreateTopicOptions struct {
	Partitions        int32
	ReplicationFactor int16
	Config            map[string]*string
}

// list topic detail
func (c *KafkaClient) ListTopic() (map[string]sarama.TopicDetail, error) {
	return c.KAdmin().ListTopics()
}

// get one topic detail
func (c *KafkaClient) GetTopic(topicName string) (meta *sarama.TopicDetail, err error) {
	topics, err := c.ListTopic()
	if err != nil {
		return
	}
	found, exists := topics[topicName]
	if exists {
		meta = &found
	} else {
		meta = nil
	}
	return
}

// topic meta data
func (c *KafkaClient) DescribeTopic(topic []string) ([]*sarama.TopicMetadata, error) {
	res, err := c.KAdmin().DescribeTopics(topic)
	if err != nil {
		return nil, err
	}
	return res, nil
}

// create topic
func (c *KafkaClient) CreateTopic(topic string, opts *CreateTopicOptions) error {
	err := c.KAdmin().CreateTopic(topic, &sarama.TopicDetail{
		NumPartitions:     opts.Partitions,
		ReplicationFactor: opts.ReplicationFactor,
		ConfigEntries:     opts.Config,
	}, false)
	if err != nil {
		return errors.New("create topic fail, err:" + err.Error())
	}
	return nil
}

// delete topic
func (c *KafkaClient) DeleteTopic(topic string) error {
	err := c.KAdmin().DeleteTopic(topic)
	if err != nil {
		return errors.New("delete topic fail, err:" + err.Error())
	}
	return nil
}

// create partitions
// --count 分片数 总和
// --assignment 指定节点分配, 默认则填nil
func (c *KafkaClient) CreatePartitions(topic string, count int32, assignment [][]int32) error {
	err := c.KAdmin().CreatePartitions(topic, count, assignment, false)
	if err != nil {
		return errors.New("create partitions fail, err:" + err.Error())
	}
	return nil
}